package com.hu.util

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

import scala.io.BufferedSource

/**
 * @Author: hujianjun
 * @Create Date: 2020/11/26 11:31
 * @Describe:
 */
class KafkaProducerUtil {
  def kafkaProducer(topicName: String, dataFilePath: String): Unit = {
    val properties = new Properties()
    properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName)
    properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName)

    //这里如果不限定泛型，则在调用send方法时会报错
    val producer = new KafkaProducer[String, String](properties)

    val buffSource: BufferedSource = io.Source.fromFile(dataFilePath)

    var recordCount = 0L
    for (line <- buffSource.getLines()) {
      val record: ProducerRecord[String, String] = new ProducerRecord[String, String](topicName, line)
      producer.send(record)
      recordCount += 1
    }

    //读取结束后关闭生产者
    producer.close()

    println(s"生产数据结束，记录数：$recordCount")
  }

}
